package com.xkcoding.mq.kafka.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Controller;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.Mapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.security.auth.callback.Callback;
import java.util.concurrent.CompletableFuture;

/**
 * @Author dengkaibing
 * @Date 2022/7/18 14:36
 */
@Controller
public class KafkaSendController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping("/send")
    @ResponseBody
    public String sendKafka() {
        for (int i = 1; i < 100; i++) {
            try {
                ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("mall-trade-product-2", "hello,kafka..." + i);
                future.addCallback(new SuccessCallback<SendResult<String, String>>() {
                    @Override
                    public void onSuccess(SendResult<String, String> stringStringSendResult) {
                        System.out.println("success to send");
                    }
                },new FailureCallback() {
                    @Override
                    public void onFailure(Throwable throwable) {
                        System.out.println("failed to send "+throwable);
                    }
                });
                CompletableFuture<SendResult<String, String>> completable = future.completable();
            }catch (Exception e) {
                return "error";
            }
        }
        return "successful";
    }
}
